package sparkCore

import org.apache.spark.{SparkConf, SparkContext}

object HelloSpark {
  def main(args: Array[String]): Unit = {
    // spark 编程
    // rdd  弹性分布式数据集   类似于 array ，操作的是数据集

    /*
    * 环境：1）  运行模式   本地模式，yarn-client，yarn-cluster  配置 本地模式 ，
    *      配置项  SparkConf ,
    *     任务  SparkConf  appName
    *     设置 内存
    *    2） SparkContext     SC  提交job任务   spark 客户端
    *
    *    3）spark 基于内存的，跑的数据如果宕机
    *    4） 依赖 spark core 依赖
    */

    //RDD  Resilient Distributed Dataset 弹性 分布式数据集
    // 弹性 spark 编程
    // 分布式  在集群上就近读取数据运行/计算
    // 数据集  集合


    // 所有操作的对象 都是一个集合 Array ，



    // 词频统计
    val conf = new SparkConf().setMaster("local[*]").setAppName("hello spark") // spark 配置
    val sc = new SparkContext(conf)  // conf 配置， sparkcontext  sc  spark 上下文

    // 构建了数据 数组
    val arr = Array("hello world","hello java","hello spark")

    val arrnew = sc.parallelize(arr)   // 将  数组转化为 rdd
    // data: ("hello world","hello java","hello spark")

    val obj = arrnew.flatMap( x => x.split(" "))  // hello world  hello java
    // flatMap    x => x.split(" ")  匿名函数 ，将每个元素空格进行切分   ((hello, world),(hello ,java),(hello, spark))
    // flatMap    切分 压平   hello, world ,hello ,java,hello, spark  一个一个
    // obj   hello, world ,hello ,java,hello, spark


    /**    1 函数方法
     *    def hello(name:string,age:int) ={
     *      函数体
     *      println("111")
     *    }
     *    2 缩写简写
     *     def hello(name:string,age:int) =   println("111")
     *    3 匿名函数
     *      (name:string,age:int) =>   println("111")
     *     4
     *      (name,age) =>   println("111")
     *      5
     *      name => println(name)
     *
     *
     */




    // val obj1 = obj.map(x => { return (x,1)} )
    val obj1 = obj.map(x =>  (x,1) )  //(hello,1)  (world,1)  (hello,1)
    // obj   hello, world ,hello ,java,hello, spark
    // map   x =>  (x,1)  匿名函数
    //map   (hello,1) (world,1) (hello,1) (java,1)
    // obj1   (key, 1)

    println("----------")
    val ss = obj1.collect()
    ss.foreach(println(_))



    val obj2 = obj1.reduceByKey((x,y) => x+y )  // (hello,3)  (world ,1)
    // 匿名函数  (x,y) => x+y   把相同的key值，value 进行相加
    //  reduceByKey 按照key进行分组
    //obj2  (hello,3) (java,1) (spark,1)

    //obj2.foreach(    x=> println(x)   )

    // foreach 循环打印
    // (hello,3)
    // (java,1)


    /**
     *   spark 编码 简洁，优雅
     *   中间的值 直接打印 没有结果
     *   spark 算子 （函数逻辑， sum ，count，avg）
     *   算子  transformation 转换类型的算子     flatMap
     *           转换类型的算子不执行结果，只执行计算的逻辑
     *
     *   算子  action  动作类型的算子
     *           执行计算结果
     */
    //简洁
    arrnew.flatMap(_.split(" ")).map((_,1)).reduceByKey((_+_)).foreach(println(_))
    // 下划线 表示 通配符  x












  }

}
